package realtime.app.dwd.db;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import realtime.util.EnvUtil;
import realtime.util.KafkaUtil;

public class DwdTradePayDetailSuc {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = EnvUtil.getSEE(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.executeSql(KafkaUtil.getTopicDbDDL("dwd_trade_pay_detail_suc_group"));
        Table orderInfo = tableEnv.sqlQuery("select " +
                " data['id'],\n" +
                " data['user_id'],\n" +
                " data['origin_amount'],\n" +
                " data['coupon_reduce'],\n" +
                " data['final_amount'],\n" +
                " data['order_status'],\n" +
                " data['out_trade_no'],\n" +
                " data['trade_body'],\n" +
                " data['session_id'],\n" +
                " data['province_id'],\n" +
                " data['create_time'],\n" +
                " data['expire_time'],\n" +
                " data['update_time']," +
                " ts" +
                " from topic_db " +
                " where `table`='order_info' and data['order_status']='1002' and `type` = 'update'");
        tableEnv.createTemporaryView("order_info",orderInfo);
        TableResult tableResult = tableEnv.executeSql("create table dwd_trade_pay_detail_suc(" +
                " id string,\n" +
                " user_id string,\n" +
                " origin_amount string,\n" +
                " coupon_reduce string,\n" +
                " final_amount string,\n" +
                " order_status string,\n" +
                " out_trade_no string,\n" +
                " trade_body string,\n" +
                " session_id string,\n" +
                " province_id string,\n" +
                " create_time string,\n" +
                " expire_time string,\n" +
                " update_time string," +
                " ts string," +
                "  primary key(id) not enforced" +
                ")" + KafkaUtil.getUpsertKafkaDDL("dwd_trade_pay_detail_suc"));
        tableEnv.executeSql("insert into dwd_trade_pay_detail_suc select * from order_info");
    }
}
